package generalObjects;

import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.Map;

import org.apache.commons.lang3.SerializationUtils;
import org.codehaus.jackson.JsonParseException;
import org.codehaus.jackson.map.JsonMappingException;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.type.JavaType;

import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.ConsistencyLevel;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.SimpleStatement;

public class CassandraHandler {

    private final Session session;

    private final ObjectMapper oMapper;

    public CassandraHandler(final String[] nodes, final String keyspace) {
        super();
        // Construct cassandra cluster.
        final Cluster cluster = Cluster.builder().addContactPoints(nodes)
                .build();
        // Connect to the given keyspace.
        session = cluster.connect(keyspace);
        oMapper = new ObjectMapper();
    }

    public Session getSession() {
        return session;
    }

    public ByteBuffer serializeObject(final Object obj) {
        // supplies a general method for serializing objects into storable blob
        // data.
        ByteBuffer bytedata = null;
        final byte[] blob = SerializationUtils.serialize((Serializable) obj);
        bytedata = ByteBuffer.wrap(blob);
        return bytedata;
    }

    public Object deSerializeObject(final ByteBuffer byteData) {
        // supplies a general method for deserializing objects into Java
        // Objects.
        final byte[] blob = new byte[byteData.remaining()];
        byteData.get(blob, 0, blob.length);

        return SerializationUtils.deserialize(blob);
    }

    public String mapToJSONString(final Map<String, Integer> obj) {
        final StringBuffer strB = new StringBuffer();
        strB.append("{");
        for (final Map.Entry<String, Integer> entry : obj.entrySet()) {
            strB.append('"' + entry.getKey() + '"' + ':' + entry.getValue()
                    + ",");
        }
        strB.deleteCharAt(strB.length() - 1);
        strB.append("}");
        return strB.toString();

    }

    public Object fromJSONString(final String jsonString, final JavaType type) {
        try {
            return oMapper.readValue(jsonString, type);
        } catch (final JsonParseException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (final JsonMappingException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (final IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        return null;
    }

    public ObjectMapper getObjectMapper() {
        return oMapper;
    }

    public ResultSet executeStatement(final String query,
            final ConsistencyLevel lvl, final Object... binds) {
        final SimpleStatement statement = new SimpleStatement(query);
        if (lvl != null) {
            statement.setConsistencyLevel(lvl);
        }
        final PreparedStatement preparedstatement = session.prepare(statement);
        return session.execute(preparedstatement.bind(binds));
    }
}
